package cn.tedu.stream.sink

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer
import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper

import java.util.Properties

/**
 * @author Amos
 * @date 2022/5/22
 */

object StreamKafkaSinkDemo {

  // 定义一个ObjectMapper对象
  private val mapper = new ObjectMapper()
  // 将student对象转换为json的字符串
  def toJsonString(o: Object): String = {
    // 将student对象转换为json的字符串
    mapper.registerModule(DefaultScalaModule)
    mapper.writeValueAsString(o)
  }

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    import org.apache.flink.api.scala._
    val source: DataStream[Student] = env.fromElements(
      Student(8, "xiaoming", "beijing biejing", "female")
    )
    // 将数据中的student转换成json的数据
    val jsonSource: DataStream[String] = source.map(stu => toJsonString(stu))

    val props = new Properties()
    props.setProperty("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092")

    val producer = new FlinkKafkaProducer[String]("test",
      new KeyedSerializationSchemaWrapper[String](new SimpleStringSchema()),
      props)
    // 自定义sink
    jsonSource.addSink(producer)
    env.execute()

  }

}

case class Student(id: Int, name: String, address: String, gender: String)
